package com.amon.amonplayer.rxjava;

import com.amon.amonplayer.rxjava.schedulder.Schedulers;

public class ObservableObserverOn<T> extends Observable<T>{
    Observable<T> source;
    Schedulers schedulers;

    public ObservableObserverOn(Observable<T> source, Schedulers schedulers) {
        this.source = source;
        this.schedulers = schedulers;
    }

    @Override
    protected void subscribeActual(Observer<T> s) {
        source.subscribe(new ObserverOnObserver<T>(s, schedulers));
    }

    private class ObserverOnObserver<T> implements Observer<T>, Runnable {
        private final Observer<T> observer;
        private final Schedulers schedulers;
        private T value;

        public ObserverOnObserver(Observer<T> s, Schedulers schedulers) {
            this.observer = s;
            this.schedulers = schedulers;
        }

        @Override
        public void onNext(T t) {
            value = t;
            schedulers.scheduleDirect(this);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }

        @Override
        public void onSubscribe() {
            observer.onSubscribe();
        }

        @Override
        public void run() {
            observer.onNext(value);
        }
    }
}
